#pragma once

#include "Socket.hpp"
#include "Reactor.hpp"
#include "Pipe.hpp"
#include "PipeAgreement.hpp"

#include <thread>

using namespace SocketModule;

class ReactorConcurrent // 高并发Reactor
{
    static const uint16_t default_port = 0;
    static const int default_wfd = -1; // 默认写端fd
    static const uint16_t default_thread_num = 5;

    typedef function<string(string &)> HandleRequset_t; // 处理接收缓冲区中的数据的方法

    void Handle(int pipe_rfd)
    {
        // 一个线程一个Reactor,把所有连接平均分配给线程们
        Reactor reactor;

        // 注册上层处理请求报文的回调函数
        reactor.Register(_handle);

        // 关心管道读端fd
        shared_ptr<Connection> pipe_r = make_shared<Pipeer>(pipe_rfd, &reactor);
        reactor.InsetConnection(pipe_rfd, pipe_r);

        reactor.Start();
    }
    void DispatchConnection() // 派发连接
    {
        uint32_t ip_client;
        uint16_t port_client;
        int sockfd = _listensocket->Accept(ip_client, port_client);
        if (sockfd < 0) // 获取新链接失败
        {
            LOG(LogLevel::ERROR) << "listensocket error";
            return;
        }
        PipeAgreement agreement;
        agreement.SetCode(Code_Type::ACCEPT);
        agreement.SetDistIp(ip_client);
        agreement.SetDistPort(port_client);
        agreement.SetFd(sockfd);

        LOG(LogLevel::DEBUG) << "新连接fd: " << sockfd;

        int n = write(_pipe_w[_pipe_w_cur], &agreement, sizeof(agreement));
        if (n > 0)
        {
            LOG(LogLevel::ERROR) << "向管道写端fd: " << _pipe_w[_pipe_w_cur] << "写入连接fd--成功--";
            _pipe_w_cur++;
            if (_pipe_w_cur >= _thread_num)
            {
                _pipe_w_cur %= _thread_num;
            }
        }
        else
        {
            LOG(LogLevel::WARNING) << "向管道写端fd: " << _pipe_w[_pipe_w_cur] << "写入连接fd--失败--";
        }
    }

public:
    ReactorConcurrent(uint16_t port = default_port, int thread_num = default_thread_num)
        : _listensocket(make_shared<TcpSocket>(port)),
          _isrunning(false),
          _thread_num(thread_num),
          _handle(nullptr),
          _pipe_w_cur(0)
    {
        // 初始化listensocket
        _listensocket->BuildSocket(SockType::LISTEN_SOCK);
    }
    bool Start()
    {
        if (_isrunning == true)
        {
            return false;
        }
        _isrunning = true;
        for (int i = 0; i < _thread_num; i++)
        {
            // 创建管道
            int fd[2];
            pipe(fd);

            // 保存管道写端
            _pipe_w.emplace_back(fd[1]);

            // 创建线程,并保存线程
            _threads.emplace_back([this](int fd)
                                  { this->Handle(fd); }, fd[0]);
        }

        while (_isrunning)
        {
            DispatchConnection(); // 派发连接
        }
        return true;
    }

    void Registered(HandleRequset_t handle)
    {
        _handle = handle;
    }
    bool Stop()
    {
        if (_isrunning == false)
        {
            return false;
        }
        _isrunning = false;

        LOG(LogLevel::ERROR) << "终止ReactorConcurrent........";

        for (int i = 0; i < _thread_num; i++)
        {
            // 创建管道报文
            PipeAgreement agreement;
            agreement.SetCode(Code_Type::STOP);

            int n = write(_pipe_w[i], &agreement, sizeof(agreement));
            if (n > 0)
            {
                LOG(LogLevel::ERROR) << "========向管道写端fd: " << _pipe_w[_pipe_w_cur] << "写入终止信号--成功=======";
                _pipe_w_cur++;
                if (_pipe_w_cur >= _thread_num)
                {
                    _pipe_w_cur %= _thread_num;
                }
            }
            else
            {
                LOG(LogLevel::WARNING) << "========向管道写端fd: " << _pipe_w[_pipe_w_cur] << "写入终止信号--失败========";
            }
        }

        // 等待所有线程退出
        for (int i = 0; i < _thread_num; i++)
        {
            _threads[i].join();
            LOG(LogLevel::ERROR) << "回收线程成功";
        }

        // 关闭所有管道写端
        for (int i = 0; i < _thread_num; i++)
        {
            close(_pipe_w[i]);
        }

        // 清空保存的管道写端
        _pipe_w.clear();
        _pipe_w_cur = 0;

        // 清空保存的线程对象
        _threads.clear();

        return true;
    }
    ~ReactorConcurrent()
    {
        _listensocket->Close();
    }

private:
    SockPtr _listensocket;
    vector<int> _pipe_w;     // 管理所有管道写端
    int _pipe_w_cur;         // 当前要被派发连接的管道写端的下标
    vector<thread> _threads; // 管理所有线程
    uint16_t _thread_num;    // 线程总数
    bool _isrunning;         // 是否运行
    HandleRequset_t _handle; // 回调上层处理请求报文的函数
};